The processing that is required to be achieved by computers has become more and more sophisticated and large-scaled. However, improvement in the performances (calculation capacity, storage volume, and the like) as hardware alone has already reached the limit. Therefore, recently, techniques of distributed processing (distributed database, distributed cache, distributed shared memory, distributed batch, and the like) with which a plurality of computers being connected mutually via a network cooperate and execute the processing in an allotted manner have especially been developed.
Recently in particular, it has become possible to acquire a RAM (main storage unit) with a capacity of 4 GB level, a hard disk (external memory device) with a capacity of 1 TB level, an NIC (network card) with a communication speed of 1 Gbps level, and the like at a low price in the field of personal computers and to connect computer devices provided with those mutually via a high-speed network easily. Therefore, it has become common to connect a plurality of such nodes (server devices) and to employ distributed processing middleware for each node to have it execute distributed processing.
The middleware is software which is typically provided between an operating system (OS) and application software and executes expansion of OS functions and multi-purpose functions of the application software. Among those, the distributed processing middleware provides functions required in the distributed processing (data synchronization among a plurality of nodes, distribution/disposition of application, start/end of process, error processing, etc.). Recently in particular, the so-called cloud computing which performs processing by a large number of nodes in a cooperated manner has become developed, so that the importance of the distributed processing middleware is especially becoming increased.
As the distributed processing methods that can be executed by a distributed processing system using the distributed processing middleware, there are a plurality of methods such as MapReduce, BSP, and MPI. Among those, a master-slave type distributed processing method is also a typical method.
The master-slave type is a kind of processing modes, with which a control component or a node called a master controls other nodes, devices, and processes called slaves. When a master-slave relation is established between nodes or between processes, control of the slave is always done by the master.
FIG. 20 is an explanatory chart showing the structure of an existing distributed processing system 900. The distributed processing system 900 is constituted by mutually connecting a plurality of computer devices via a network 930. One of the computer devices is a master node 910, and the others are slave nodes 920a, 920b, - - - (referred to as slave node 920 as a general term).
Distributed processing middleware 912 is installed on an OS 911 (operating system) of the master node 910. Similarly, distributed processing middleware 922 is installed on an OS 921 of the slave node 920.
The master node 910 distributes a distributed application to the slave node 920 to have it execute necessary processing. However, the part processed by the slave node is not completely the same as that of the master node but may often be a part of the processing or a part of the data. Through executing different processing or processing different data in each of the nodes, a plurality of kinds of processing can be distributed and parallelized. Therefore, the processing can be executed at a high speed.
FIG. 21 is an explanatory chart showing the structure of a distributed application 950 that is executed by the distributed processing system 900 shown in FIG. 20. The distributed application 950 is constituted by combining a plurality of sets of processing modules, common modules, setting files, binary data (image file and the like), etc.
Each of the modules is an execution format file, a common library file, or the like. In a case of Windows (registered trademark), for example, it is an EXE file, a DLL file, or the like. In a case of Linux (registered trademark), it is a shell script, an execution file, a library file, or the like. In a case of Java (registered trademark), it is a Jar file, a Class file, or the like.
In any cases, each of the modules is constituted with “metadata” and “execution region”. The metadata is data regarding the module such as the module name, signature (a set of method name and argument), field, constant, and the like. The execution region is an execution code such as the method which is actually operated.
Further, each module includes initialization processing at the time of execution. More specifically, DLL initialization of Windows is DllMain( ), library initialization of Linux is _init( ), and initialization of Java is <clinit>, <init>, for example.
When the distributed application 950 starts up, metadata of the processing module started first in each node is loaded to analyze the metadata, and the constants and execution regions are disposed in a memory to be able to actually execute the processing. If a dependent relation for another module is included therein, actions of loading that module and solving the dependent relation to be in a state where call between the modules can be done (called “link”) are repeated.
Normally, each module has a part for the initialization processing. Through executing the initialization processing after being linked, the concerned application comes to be in an executable state. That is, through repeating actions of “reading out necessary file (Load)→solving dependent relation of each of the modules (Link)→execute initialization processing (Initialize)”, the distributed application 950 becomes executable.
However, the loading timing of each of the modules depends on the implementation of each device, and there are a case where the necessary modules are all loaded at the time of startup and a case where the modules are not loaded until those are actually referred.
As the related technical documents thereof, there are following documents. Depicted in Patent Document 2 is a distributed processing system which selects object data based on a set rule and improves the efficiency of the processing that uses it. Depicted in Patent Document 2 is a technique which performs management of distributed processing by using graphic user interface. Depicted in Patent Document 3 is a technique regarding disposition of metadata on a network.
Depicted in Patent Document 4 is a distributed processing system which uses a processing device that is provided with a sub-processor specialized in specific processing. Depicted in Patent Document 5 is deployment of web services on a network. Depicted in Patent Document 6 is a technique which acquires information required in a distributed processing device from other devices if the necessary information is not provided in the own device to share the information among all the devices.
Depicted in Patent Document 7 is a technique regarding layout of metadata of multilayer software. Depicted in Non-Patent Document 1 is the outline of Hadoop that is typical distributed processing middleware.    Patent Document 1: Japanese Unexamined Patent Publication 2012-063832    Patent Document 2: Japanese Unexamined Patent Publication 2011-034137    Patent Document 3: Japanese Unexamined Patent Publication 2008-171277    Patent Document 4: Japanese Unexamined Patent Publication 2006-155187    Patent Document 5: Japanese Unexamined Patent Publication 2004-533687    Patent Document 6: Japanese Unexamined Patent Publication 2004-342042    Patent Document 7: Japanese Unexamined Patent Publication 2003-505750    Non-Patent Document 1: “Let's Try Distributed Processing” Takeshi Wakimoto. Sep. 29, 2010, (searched on May 14, 2013), Pasona TECH, <URL: http://www.pasonatech.co.jp/techlab/rdtrend/rep7.jsp>
In the current Description, the distributed processing system 900 which operates in a Java environment such as Hadoop depicted in Non-Patent Document 1 described above is assumed, so that it is to be noted that each of processing modules and common modules are written as class (.class) files. Note, however, that the present invention is not necessarily limited to the Java environments.
The distributed application 950 shown in FIG. 21 includes processing modules 951 to 953 (step01.class to step03.class) of the first to third steps, first to third common modules 954 to 956 (shared01.class to shared03.class) used in each of the steps, other setting files, binary data, and the like.
Further, the distributed application 950 also includes a processing module 952a (step02_slave.class) on the slave node side of the second step since the master node 910 and each of the slave nodes 920 executes the second step by performing distributed processing in a cooperative manner while the first and the third steps are executed by the master node 910 alone. Each of the above-described files is put into one file under a file name such as “distributed batch application 1.jar) or the like.
FIG. 22 is an explanatory chart showing actions of a case where the distributed application 950 shown in FIG. 21 is executed by the existing distributed processing system 900 shown in FIG. 20. The master node 910 upon receiving an execution command from a user distributes the file (distributed batch application 1.jar) of the distributed application 950 to each of the slave nodes 920. In each of the nodes, each of the distributed processing middleware 912 (922) starts up a process, expands the file of the distributed application 950, loads the class file, and executes it.
However, as shown in FIG. 21, not all of the steps of the distributed application 950 execute the distributed processing. Thus, if simply the file (distributed batch application 1.jar) of the distributed application 950 is distributed to each node, the file that is not used in the distributed processing is also transferred. This is clearly useless transfer. Especially, when a file of large size such as images, moving pictures, audios, or the like is contained, it is wasted unless such file is used.
Further, when the file is transferred at the time of execution, an issue of delay in the startup processing of the slave node 920 occurs. Referring to a case shown in FIG. 21, distributed processing is performed in the second step. When the master node 910 distributes the file of the distributed application 950 to each of the slave nodes 920 at a point where the first step is completed and the second step is started, delay is generated in the distribution processing and in the processing executed at a stage of expanding the “distributed batch application 1.jar” file and loading the class file.
In order for not causing the delay, there is considered a method with which the file of the distributed application 950 is distributed and expansion and loading of the class file are to be completed on each of the slave nodes 920 side while the master node 910 alone is executing the first step. However, this may cause another issue.
In the distributed processing, the number of the slave nodes 920 may often become several thousands, several tens of thousands, or may be in a scale that is greater than those. Further, the distributed processing middleware 912 (922) normally allots the processing to the set number of slave nodes after checking the condition of the resources (calculator resources) of each node at a point where the distribution processing is to be executed actually. That is, it is not possible to know which of the slave nodes executes the distributed processing until that point.
Therefore, if the file of the distributed application 950 is distributed unconditionally to each of the slave nodes 920 in advance, the file may be distributed in advance even to the slave node 920 to which the processing is actually not allotted. This is also useless transfer. That is, to distribute the file of the distributed application 950 in advance for not causing delay in the startup processing of the slave nodes 920 is not considered appropriate.
Further, the distributed processing executed by each of the nodes (the master node 910 and each of the slave nodes 920) in a cooperative manner in many cases is executed to perform same processing on different target data, so that the startup processing executed at the time of startup in each node is often the common processing.
Regardless of that, the distributed processing middleware 922 of each of the slave nodes 920 normally generates and deletes the process by each of the operation steps. Thus, each of the nodes executes the same startup processing every time for each of the operation steps. This is also useless processing that may be a reason for causing delay.
A technique capable of overcoming the above-described issues and achieving both reducing the data volume to be transferred to each of the slave nodes at the time of execution and starting the distributed processing at a high speed is not depicted in Patent Documents 1 to 7 and Non-Patent Document 1 described above.
It is therefore an object of the present invention to provide a distributed processing system, a distributed processing device, a distributed processing method, and a distributed processing program capable of reducing the data volume to be transferred to each of the slave nodes at the time of execution and starting the distributed processing at a high speed.